package com.shujia.streaming

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Code05Transform {
  def main(args: Array[String]): Unit = {

    // 如果要使用SparkStreaming 需要获取其操作对象 => StreamingContext
    // 如果创建StreamingContext对象那么需要SparkContext对象
    // local[2] 表示本地模式 并且设置并发数为2 此项为必须项
    val sc = new SparkContext(new SparkConf().setMaster("local[2]").setAppName("wordCountStreaming"))

    // Durations.seconds(5) 表示每5秒启动一个计算任务
    val streamingContext: StreamingContext = new StreamingContext(sc, Durations.seconds(5))
    // 可以监听socket端口获取数据 master:8888 在Linux中可以通过nc -lk master 8888 启动
    val stream: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 8888)


    // 对于transform来说，参数是RDD返回值也是一个RDD
    // 和foreachRDD相比区别为是否存在RDD作为返回值
    stream
      .transform(
        (rdd: RDD[String]) => {
          rdd
            .flatMap((line: String) => line.split(","))
            .map((word: String) => (word, 1))
        }
      )
      .reduceByKey((x: Int, y: Int) => x + y)
      .print()

    // 需要启动 streamingContext
    streamingContext.start()
    // 等待停止信号 <- 手动停止
    streamingContext.awaitTermination()

    // Streaming程序是 7 * 24 小时不间断运行，那么不需停止
    //    streamingContext.stop()

  }
}
